package edu.purdue.cs505;

import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;

public class FIFOReliableBroadcastImpl implements FIFOReliableBroadcast, BroadcastReceiver
{
  public static boolean srbOn = false; // true if SRB is on
  public static long deliveryDelay = 0; // delay (ms) for delivery of messages on receiver side

  private ReliableBroadcast reliableBroadcast;
  private Map<String, Integer> nextSequence;
  private Map<String, Set<Integer>> obsSequence;
  private Queue<DelayedMessage> messageBag;
  private BroadcastReceiver callback;
  private boolean halted;
  private Process currentProcess;
  private int selfSeqNum;

  public FIFOReliableBroadcastImpl()
  {
    reliableBroadcast = new ReliableBroadcastImpl();
    nextSequence = new ConcurrentHashMap<String, Integer>();
    obsSequence = new ConcurrentHashMap<String, Set<Integer>>();
    messageBag = new PriorityBlockingQueue<DelayedMessage>();
    callback = null;
    halted = false;
    selfSeqNum = 1;
  }

  @Override
  public void init(Process currentProcess)
  {
    this.currentProcess = currentProcess;
    nextSequence.put(currentProcess.getProcessID(), 1);
    obsSequence.put(
        currentProcess.getProcessID(),
        Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>()));
    reliableBroadcast.init(currentProcess);
    reliableBroadcast.rblisten(this);
    if (srbOn) {
      SRBDeliveryThread deliverThread = new SRBDeliveryThread();
      new Thread(deliverThread).start();
    }
  }

  @Override
  public void addProcess(Process p)
  {
    nextSequence.put(p.getProcessID(), 1); // we expect 1 to be the first sequence number
    obsSequence.put(
        p.getProcessID(),
        Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>()));
    reliableBroadcast.addProcess(p);
  }

  @Override
  public void rbroadcast(Message m)
  {
    reliableBroadcast.rbroadcast(m);
  }

  @Override
  public void rblisten(BroadcastReceiver m)
  {
    callback = m;
  }

  @Override
  public void halt()
  {
    halted = true;
    reliableBroadcast.halt();
  }

  @Override
  public synchronized void rdeliver(Message m)
  {
    //System.out.println("deliver " + m);
    String sender = m.getProcessID();

    // just deliver if we originally sent the message
    if (sender.equals(currentProcess.getProcessID()) && !srbOn) {
      this.callback.rdeliver(m);
      return;
    }

    // add this message to the bag
    long deliveryTime = srbOn ? new Date().getTime() + deliveryDelay * 1000 : m.getMessageNumber();

    // make sure the seq num is bigger than the biggest we've delivered
    if (m.getMessageNumber() >= nextSequence.get(m.getProcessID())
        || sender.equals(currentProcess.getProcessID())) {
      if (sender.equals(currentProcess.getProcessID())) {
        // need to set the seq num since sending to myself never sets it
        m.setMessageNumber(selfSeqNum++);
      }
      DelayedMessage dm = new DelayedMessage(m, deliveryTime);
      //System.out.println("adding " + dm);
      messageBag.add(dm);
    }

    if (srbOn) {
      // get rid of obsolete messages
      Set<DelayedMessage> obsoleteMessages = new HashSet<DelayedMessage>();

      // build a list of obsolete messages by iterating through the bag
      for (DelayedMessage dm : messageBag) {
        if (dm.message.getProcessID().equals(sender)) {
          for (int i : m.getObsoletedMessages()) {
            if (dm.message.getMessageNumber() == i) {
              //System.out.println("obs: " + dm.message);
              obsoleteMessages.add(dm);
              obsSequence.get(dm.message.getProcessID()).add(dm.message.getMessageNumber());
            }
          }
        }
      }

      // remove all the messages
      if(messageBag.removeAll(obsoleteMessages)) {
        //System.out.println("removed obs messages");
      }
    }
    else {
      // we only do this in non-SRB mode since the delivery thread handles all deliveries for SRB
      // check the bag for deliverable messages until there are no deliverable messages
      while (!messageBag.isEmpty()) {
        boolean found = false;
        int nextSeq = nextSequence.get(sender);
        DelayedMessage foundMessage = null;
        for (DelayedMessage dm : messageBag) {
          Message currentMessage = dm.message;
          if (currentMessage.messageNumber == nextSeq
              && currentMessage.getProcessID().equals(sender)) {
            found = true;
            //System.out.println("Delivering " + Integer.toString(nextSeq));
            foundMessage = dm;
            nextSequence.put(sender, ++nextSeq);
            callback.rdeliver(currentMessage);
            break; // we must restart the iteration since the list size changed
          }
        }
        if (found) { messageBag.remove(foundMessage); }
        if (!found) break;
      }
    }
  }

  private class SRBDeliveryThread implements Runnable
  {
    @Override
    public void run()
    {
      while (!halted) {
        if (!messageBag.isEmpty()) {
          // System.out.println("bag: " + messageBag.size());
          for (DelayedMessage dm : messageBag) {
            // iterate messages in order
            if (dm.deliveryTime > new Date().getTime()) break;

            // message is due for delivery, so check FIFO order
            Message currentMessage = dm.message;
            int nextSeq = nextSequence.get(dm.message.getProcessID());
            while (obsSequence.get(currentMessage.getProcessID()).contains(nextSeq)) {
              obsSequence.get(currentMessage.getProcessID()).remove(new Integer(nextSeq));
              //System.out.println("skipped obs seq " + nextSeq);
              ++nextSeq;
              nextSequence.put(dm.message.getProcessID(), nextSeq);
            }
            //if(new Date().getTime() % 10000 == 0) System.out.println("next seq: " + nextSeq);
            if (currentMessage.messageNumber == nextSeq) {
              messageBag.remove(dm); // TODO: this may not work due to modifying in a for each
              ++nextSeq;
              nextSequence.put(dm.message.getProcessID(), nextSeq);
              callback.rdeliver(currentMessage);
              break; // we must restart the iteration since the list size changed
            }
          }
        }
      }
    }
  }
}
